package cn.doitedu.day05

import cn.doitedu.day01.utils.SparkUtil
import org.apache.spark.rdd.RDD

/**
 * @Date 22.4.3
 * @Created by HANGGE
 * @Description
 */
object C21_转换算子_CoGroup {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc
    val arr1 = Array(("a", 1), ("c", 22), ("d", 33), ("d", 333), ("a", 2), ("b", 222), ("f", 2))
    val arr2 = Array(("a", 1111), ("c", 2222), ("d", 3333), ("o", 333), ("a", 111), ("b", 422), ("f", 21),("e", 21))

    val rdd1 = sc.parallelize(arr1, 1)
    val rdd2 = sc.parallelize(arr2, 1)
    /**
     * 将所有数据  按照key进行分组
     * 元组_2的第一个迭代器是  第一个rdd的分组后的 v
     * 元组_2的第二个迭代器是  第二个rdd的分组后的 v
     */
    val res: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
    res.foreach(tp=>println(tp._1))
    res.foreach(tp=>println((tp._1,tp._2._1.toList ++ tp._2._2.toList)))

  }

}
